package com.hitqz.robot.driver.zmq.protocol.component;

import lombok.extern.slf4j.Slf4j;
import org.zeromq.*;

import java.util.LinkedList;

/**
 * 基于ZeroMQ 的路由线程，用于处理前端客户端和后端工作进程之间的消息传递
 */
@Slf4j
public class RouterThread implements Runnable {

    /**
     * 前端端口，用于绑定前端套接字
     * */
    public int frontPort;

    /**
     * 后端名称，用于绑定后端套接字
     * */
    private final String backName;

    /**
    * 启动监听器名字
    * */
    private String moniterName;


    public LinkedList<ZFrame> workers = new LinkedList<>();

    private final LinkedList<ZMsg> requests = new LinkedList<>();

    public RouterThread(int frontPort, String backName) {
        this.frontPort = frontPort;
        this.backName =backName;
    }


    @Override
    public void run() {

        try(ZContext ctx = new ZContext()) {

            ctx.setSndHWM(0);
            ctx.setRcvHWM(0);

            ZMQ.Socket frontend = ctx.createSocket(SocketType.ROUTER);
            ZMQ.Socket backend = ctx.createSocket(SocketType.ROUTER);

            /*
             * 参数设置
             * */
            frontend.setRcvHWM(0);
            frontend.setSndHWM(0);
            frontend.setRouterHandover(true);

            frontend.setTCPKeepAlive(1);
            frontend.setTCPKeepAliveCount(3);
            frontend.setTCPKeepAliveIdle(60L);

            frontend.setTCPKeepAliveInterval(10L);
            frontend.setSendBufferSize(1024* 10);
            frontend.setReceiveBufferSize(1024* 10);

            /*
             * 参数设置
             * */
            backend.setRcvHWM(0);
            backend.setSndHWM(0);
            backend.setRouterHandover(true);
            backend.setSendBufferSize(1024* 10);
            backend.setReceiveBufferSize(1024* 10);

            backend.setTCPKeepAlive(1);
            backend.setTCPKeepAliveCount(3);
            backend.setTCPKeepAliveIdle(60L);
            backend.setTCPKeepAliveInterval(10L);


            /**
             * 绑定端口
             * */
            boolean bindFront = frontend.bind("tcp://*:" + frontPort);

            log.info("[wheel][zmq] 绑定:{},绑定结果 {}", frontPort, bindFront);

            /**
             * 绑定端口
             * */
            boolean bindBack = backend.bind("ipc://" + backName);

            log.info("[wheel][zmq] 绑定:{}, 绑定结果 {}", backName, bindBack);

            // 创建 pollItemsss
            ZMQ.PollItem fitem = new ZMQ.PollItem(frontend, ZMQ.Poller.POLLIN);
            ZMQ.PollItem bitem = new ZMQ.PollItem(backend, ZMQ.Poller.POLLIN);

            ZMQ.Poller poller = ctx.createPoller(2);

            poller.register(fitem);
            poller.register(bitem);

            while (!Thread.currentThread().isInterrupted()) {
                poller.poll();
                try {
                    if (fitem.isReadable()) {
                        ZMsg msg = ZMsg.recvMsg(fitem.getSocket());
                        ZFrame firstMsg = msg.getFirst();
                        if (firstMsg.toString().startsWith("00")) {
                            msg.unwrap();
                            firstMsg = msg.getFirst();
                        }
                        if (new String(firstMsg.getData()).equals("clientSendMsg")) {
                            msg.pop();
                            msg.send(frontend);
                        } else {
                            this.requests.add(msg);
                        }
                    }
                } catch (Exception e ){
                    log.error("[wheel][zmq]front解析错误:" + e);
                }

                try {
                    if (bitem.isReadable()) {
                        // worker 发送消息过来
                        ZMsg msg = ZMsg.recvMsg(bitem.getSocket());
                        ZFrame readyOrAddress = msg.getLast();
                        if (new String(readyOrAddress.getData()).equals("ready")) {
                            ZFrame workerID = msg.unwrap();
                            this.workers.addLast(workerID);
                            msg.destroy();
                        } else {
                            msg.pop();
                            msg.send(frontend);
                        }
                    }
                } catch (Exception e) {
                    log.error("[wheel][zmq]backend解析错误:" + e);
                }

                while (!this.workers.isEmpty() && !this.requests.isEmpty()) {
                    ZMsg request = this.requests.removeFirst();
                    ZFrame worker = this.workers.removeFirst();
                    request.wrap(worker);
                    request.send(backend);
                }
            }

            frontend.close();
            backend.close();

        } catch (Exception e) {
            log.error("[wheel][zmq]routerThread异常退出:",e);
            Thread router = new Thread(new RouterThread(frontPort, backName));
            router.start();
        }
    }
}
